package cn.gwm.flink.streaming.dwd.check;

import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.dwd.model.DelayEnum;
import cn.hutool.json.JSONObject;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;

/**
 * 当前数据时间在上一条之前，则过滤掉
 * 当前数据时间超过当前时间[误差10分钟吧]，则过滤掉
 */
public class LateDataFunction extends RichMapFunction<JSONObject, JSONObject> {
    private ValueState<Long> preTimeState;

    private static final Long ERROR_TIME = 10 * 60 * 1000L;

    @Override
    public void open(Configuration parameters) throws Exception {
        preTimeState = getRuntimeContext().getState(new ValueStateDescriptor<>("preTimeState", Long.class));
    }

    @Override
    public JSONObject map(JSONObject entries) throws Exception {
        if (entries.getLong(BaseFields.tid) > (System.currentTimeMillis() + ERROR_TIME)) {
            entries.set(BaseFields.DELAY, DelayEnum.FUTURE.getCode());
            return entries;
        }

        Long preTime = preTimeState.value();
        if (preTime == null) {
            preTimeState.update(entries.getLong(BaseFields.tid));
            entries.set(BaseFields.DELAY, DelayEnum.NORMAL.getCode());
            return entries;
        } else {
            if (preTime >= entries.getLong(BaseFields.tid)) {
                entries.set(BaseFields.DELAY, DelayEnum.DELAY.getCode());
                return entries;
            } else {
                preTimeState.update(entries.getLong(BaseFields.tid));
                entries.set(BaseFields.DELAY, DelayEnum.NORMAL.getCode());
                return entries;
            }
        }
    }
}
